package org.example.sync;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.util.Random;

/**
 * 生产者单个确认
 */
public class TopicsProducerConfirmBatchTest {


    public static final String CONFIRM_EXCHANGE = "producer_confirm_individually";


    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            //声明广播类型的交换机
            channel.exchangeDeclare(CONFIRM_EXCHANGE, BuiltinExchangeType.DIRECT);
            //Publisher confirms are a RabbitMQ extension to the AMQP 0.9.1 protocol,
            // so they are not enabled by default.
            // Publisher confirms are enabled at the channel level with the confirmSelect method:
            channel.confirmSelect();
            //声明一个交换机
            long start = System.currentTimeMillis();
            System.out.println("开始确认时间:" + start);
            int batchCount = 100;
            int count = 0;
            for (int i = 0; i < 1000; i++) {
                String randomMessage = new Random(10).nextInt() + "";
                channel.basicPublish(CONFIRM_EXCHANGE, "", null, ("消息" + randomMessage).getBytes());
                //等待5s确认时间
                count++;
                if (i % batchCount == 0) {
                    channel.waitForConfirmsOrDie(5_000);
                    count = 0;
                }
            }
            if (count > 0) {
                channel.waitForConfirmsOrDie(5_000);
            }
            System.out.println("确认结束：" + (System.currentTimeMillis() - start) + "ms");
            // In the Java client, when we supply no parameters to queueDeclare() we
            // create a non-durable, exclusive, autodelete queue with a generated name:
            // AMQP.Queue.DeclareOk declareOk = channel.queueDeclare();
            // 获取一个非持久化，排他，自动删除的队列
            // String queueName = declareOk.getQueue();
            // channel.queueBind(queueName,PUB_EXCHANGE,"");
        }
    }

    private static String getRoutingKey(String[] args) {
        if (args.length < 1)
            return "anonymous.info";
        return args[0];
    }
}
